package spark

import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.sql.{Connection, DriverManager, PreparedStatement}

object ValidInvalidCount1 {
  // 杨茗茹
  def main(args: Array[String]): Unit = {

    //初始化 Spark 和 Streaming 上下文
    val conf = new SparkConf().setAppName("ValidInvalidCount").setMaster("local[*]")
    //设置应用名称为 "ValidInvalidCount"，并在本地模式下运行
    val ssc = new StreamingContext(conf, Seconds(8))
    ssc.sparkContext.setLogLevel("WARN")

    //配置 Kafka 连接参数
    val kafkaParams = Map(
      "bootstrap.servers" -> "192.168.136.128:9092",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "group.id" -> "niit",
      "auto.offset.reset" -> "earliest"
    )//连接到 Kafka 集群的指定地址
     // 使用 "niit" 作为消费者组 ID
    //从最早的消息开始消费

    // 创建 Kafka 数据流
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Array("orders"), kafkaParams)//订阅名为 "orders" 的主题
    )

    // 数据处理与统计
    // 提取isValid字段并映射为(isValid, 1)对
    val validInvalidPairs = stream.map(record => {
      val fields = record.value().split("\t")
      val isValid = fields(6)//从每条消息的第 7 个字段（索引 6）提取有效性标识
      val isValidBoolean = isValid match {
        case "Y" => true
        case "N" => false
        case _ => false // 默认无效将 "Y" 映射为 true，"N" 和其他值映射为 false
      }
      (isValidBoolean, 1L)
    })

    // 按isValid分组并统计数量
    val countStream = validInvalidPairs.reduceByKey(_ + _)

    //结果输出与持久化
    countStream.print()//输出到控制台
    // 将结果写入MySQL
    countStream.foreachRDD { rdd => //采用RDD转换链处理数据流
      if (!rdd.isEmpty()) {
        rdd.foreachPartition { partition =>
          var conn: Connection = null
          var stmt: PreparedStatement = null
          try {
            // 创建数据库连接 - 不再显式加载驱动，让JDBC自动发现
            val url = "jdbc:mysql://47.93.166.116:3306/final3?useSSL=false&serverTimezone=UTC"
            val user = "root"
            val password = "Root123!"
            conn = DriverManager.getConnection(url, user, password)

            // 批量插入数据
            conn.setAutoCommit(false) // 启用事务处理，提高效率
            partition.foreach { case (isValid, count) =>
              val sql = "INSERT INTO task1 (is_valid, count) VALUES (?, ?)"//task1表
              stmt = conn.prepareStatement(sql)
              stmt.setBoolean(1, isValid)
              stmt.setLong(2, count)
              stmt.executeUpdate()
              stmt.close() // 关闭Statement
            }
            conn.commit() // 提交事务
          } catch {
            case e: Exception =>
              e.printStackTrace()
              if (conn != null) conn.rollback() // 发生异常时回滚事务
          } finally {
            // 确保资源关闭
            if (stmt != null) try { stmt.close() } catch { case e: Exception => e.printStackTrace() }
            if (conn != null) try { conn.close() } catch { case e: Exception => e.printStackTrace() }
          }
        }
      }
    }
    //启动流处理
    ssc.start()
    ssc.awaitTermination()
  }
}